/*
 ***************************************************************************************
 * All rights Reserved, Designed By RTF
 * @Title:  KafkaManager.java   
 * @Package com.rtf.repair.mq.kafka.producer
 * @Description: kafka 生产者管理类，管理其生命周期  
 * @author: 司福林
 * @date:   2020-10-17 16:39:00   
 * @version V1.0 
 * @Copyright: 2020 RTF. All rights reserved.
 * 注意：本内容仅限于公司内部使用，禁止外泄以及用于其他的商业目
 *  ---------------------------------------------------------------------------------- 
 * 文件修改记录
 *     文件版本：         修改人：             修改原因：
 ***************************************************************************************
 */
package com.rtf.framework.mq.kafka.producer;

import java.util.Map;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.util.Assert;

import com.rtf.framework.mq.common.Service;
import com.google.common.collect.Maps;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class KafkaManager implements Service {

	private KafkaProducerProperties properties;

	private Producer<String, byte[]> producer;

	/**
	 * Instantiates a new rocket mq manager.
	 *
	 * @param properties the properties
	 */
	public KafkaManager(KafkaProducerProperties properties) {
		Assert.notNull(properties, "KafkaProducerProperties is null ");

		this.properties = properties;
	}

	@Override
	public void init() {
		Map<String, Object> conf = Maps.newHashMap();
		conf.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getServers());
		conf.put(ProducerConfig.CLIENT_ID_CONFIG, properties.getClientName());
		conf.put(ProducerConfig.RETRIES_CONFIG, properties.getRetries());
		conf.put(ProducerConfig.BATCH_SIZE_CONFIG, properties.getBatchSize());
		conf.put(ProducerConfig.LINGER_MS_CONFIG, properties.getLinger());
		conf.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, properties.getMaxRequestSize());
		conf.put(ProducerConfig.BUFFER_MEMORY_CONFIG, properties.getBufferMemory());
		conf.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, properties.getRequestTimeout());
		conf.put(ProducerConfig.ACKS_CONFIG, properties.getAcks());
		conf.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, properties.getCompressionType());
		conf.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, properties.getPartitionerClass());
		producer = new KafkaProducer<String, byte[]>(conf, new StringSerializer(), new ByteArraySerializer());
	}

	@Override
	public void start() {

	}

	@Override
	public void destroy() {
		producer.close();

	}

	public Producer<String, byte[]> getProducer() {
		return producer;
	}

}
